package com.youkol.iot.demo.emqx.client;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import lombok.extern.slf4j.Slf4j;

/**
 * Mqtt3SubMessageListener
 */
@Slf4j
class Mqtt3SubMessageListener implements IMqttMessageListener {

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.debug("sub topic: " + topic);
        log.debug("sub payload: " + message.getPayload());

        // TODO 这里可以执行消息桥接转发操作
    }

}

/**
 * Mqtt3MessageCallback
 */
@Slf4j
class Mqtt3MessageCallback implements MqttCallback {

    @Override
    public void connectionLost(Throwable cause) { // 连接丢失后，一般在这里面进行重连
        log.debug("连接断开，可以做重连");

    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // subscribe后得到的消息会执行到这里
        log.debug("topic: " + topic);
        log.debug("Qos: " + message.getQos());
        log.debug("payload: " + new String(message.getPayload()));

        // TODO 这里可以执行消息桥接转发操作
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("deliveryComplete: " + token.isComplete());

    }


}

/**
 * MQTT3.0 协议的设备接入示例
 *
 * @author jackiea
 */
@Slf4j
public class Mqtt3Client {

    public void connect() {
        String broker = "tcp://127.0.0.1:1883";
        String clientId = "mqtt_dev_jackiea_" + System.currentTimeMillis();
        String username = "mqtt_dev";
        String password = "wkrj@123";

        try (
            // Paho Mqtt 客户端
            MqttClient client = new MqttClient(broker, clientId, new MemoryPersistence());
        ) {
             // Paho Mqtt 连接参数
             MqttConnectOptions options = new MqttConnectOptions();
             options.setCleanSession(true); // 保留会话
             options.setKeepAliveInterval(180);
             options.setUserName(username);
             options.setPassword(password.toCharArray());

             // 设置回调
            //  client.setCallback(new Mqtt3MessageCallback());

             // 建立连接
             client.connect(options);
             log.debug("borker: {} connected.", broker);

             // Paho Mqtt 消息订阅
             String topicSub = "/dev/topic/sub";
             client.subscribe(topicSub, new Mqtt3SubMessageListener());
            // client.subscribe(topicSub);
             log.debug("subscribe: {}", topicSub);

             // Paho Mqtt 消息发布
             String pubTopic = "/dev/topic/pub";
             String content = "{\"msg\": \"hello world\"}";
             MqttMessage message = new MqttMessage(content.getBytes());
             message.setQos(0);
             client.publish(pubTopic, message);

             Thread.sleep(5000);
        } catch (Exception ex) {
            log.error("error occur.", ex);
        }

    }

    public static void main(String[] args) {
        Mqtt3Client client = new Mqtt3Client();
        client.connect();
    }
}
